package org.budo.canal.message.handler;

import java.util.List;

import org.budo.support.lang.util.ListUtil;
import org.budo.support.lang.util.StringUtil;
import org.budo.support.lang.util.ThreadUtil;

import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.Header;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;

import lombok.extern.slf4j.Slf4j;

/**
 * @author limingwei
 */
@Slf4j
public class DefaultCanalMessageHandler implements CanalMessageHandler {
    @Override
    public void handleMessage(Message message, ClientIdentity clientIdentity) {
        List<Entry> entries = message.getEntries();

        if (null == entries || entries.isEmpty()) {
            log.debug("#33 entries=" + entries + ", message=" + message + ", clientIdentity=" + clientIdentity);
            return;
        }

        for (Entry entry : entries) {
            this.handleEntry(entry);
        }

        if (-1 == message.getId() || 0 == entries.size()) {
            ThreadUtil.sleep(100);
        }
    }

    protected void handleEntry(Entry entry) {
        Header header = entry.getHeader();

        String schemaName = header.getSchemaName();
        String tableName = header.getTableName();

        if (StringUtil.isEmpty(schemaName) || StringUtil.isEmpty(tableName)) {
            this.logEntryInfo(entry);
            return;
        }

        this.handleCanalEntry(entry);
    }

    protected void handleCanalEntry(Entry entry) {
        this.logEntryInfo(entry);
    }

    private void logEntryInfo(Entry entry) {
        Header header = entry.getHeader();

        String schemaName = header.getSchemaName();
        String tableName = header.getTableName();

        long serverId = header.getServerId(); // 可用于双向同步
        String logfileName = header.getLogfileName();
        long logfileOffset = header.getLogfileOffset();

        RowChange rowChange = RowChangeUtil.rowChangeParseFromByteString(entry.getStoreValue());
        EventType eventType = rowChange.getEventType();

        if (EventType.QUERY == eventType) {
            log.info("#42 QUERY, this=" + this + ", serverId=" + serverId + ", table=" + schemaName + "." + tableName + ", logfileName=" + logfileName + ", logfileOffset=" + logfileOffset //
                    + ", rowChange=" + RowChangeUtil.rowChangeToString(rowChange));
            return;
        }

        // #42 QUERY, serverId=1, table=`warehouse_1`., logfileName=mysql-bin.000004,
        // logfileOffset=652, rowChange=rowChange.sql=CREATE DATABASE `warehouse_1`
        // CHARACTER SET utf8 COLLATE utf8_general_ci

        if (EventType.INSERT == eventType) {
            log.info("#48 INSERT, this=" + this + ", serverId=" + serverId + ", table=" + schemaName + "." + tableName + ", logfileName=" + logfileName + ", logfileOffset=" + logfileOffset //
                    + ", rowChange=" + RowChangeUtil.rowChangeToString(rowChange));
            return;
        }

        if (EventType.DELETE == eventType) {
            log.info("#54 DELETE, this=" + this + ", table=" + schemaName + "." + tableName + ", logfileName=" + logfileName + ", logfileOffset=" + logfileOffset //
                    + ", rowChange=" + RowChangeUtil.rowChangeToString(rowChange));
            return;
        }

        if (EventType.UPDATE == eventType && !ListUtil.isNullOrEmpty(rowChange.getRowDatasList())) {
            log.info("#62 UPDATE, this=" + this + ", serverId=" + serverId + ", table=" + schemaName + "." + tableName + ", logfileName=" + logfileName + ", logfileOffset=" + logfileOffset //
                    + ", rowChange=" + RowChangeUtil.rowChangeToString(rowChange));
            return;
        }

        // rowChange.rowDatas is null
        if (EventType.UPDATE == eventType) {
            log.debug("#69 UPDATE, this=" + this + ", table=" + schemaName + "." + tableName + ", logfileName=" + logfileName + ", logfileOffset=" + logfileOffset //
                    + ", rowChange=" + RowChangeUtil.rowChangeToString(rowChange));
            return;
        }

        log.warn("#65 eventType=" + eventType + ", this=" + this + ", serverId=" + serverId + ", table=" + schemaName + "." + tableName + ", logfileName=" + logfileName + ", logfileOffset=" + logfileOffset);
    }
}